package app.dwd;


import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

import java.time.Duration;

/**
 *
 */
public class Dwd_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_DwdTradePayDetailSuc().init(
                3011,
                2,
                "Dwd_DwdTradePayDetailSuc"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_DwdTradeOrderDetail");
        // 2. 读取字典表
//        readBaseDic(tEnv);tEnv
        // 3. 从 ods_db 过滤出 支付成功表数据
        Table paymentInfo = tEnv.sqlQuery("select " +
                "data['order_id'] order_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                " ts, " +
                " current_row_timestamp() row_op_ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info' " +
                "and `type`='insert' " +
                "and `data`['payment_status']='1602' ");
        tEnv.createTemporaryView("payment_info", paymentInfo);

        //paymentInfo.execute().print();

        Table order_info = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' ");
        tEnv.createTemporaryView("order_info", order_info);
//        order_info.execute().print();
//
        Table paySuc = tEnv.sqlQuery("SELECT user_id,\n" +
                "order_id,\n" +
                "payment_type,\n" +
                "callback_time,\n" +
                "origin_amount,\n" +
                "coupon_reduce,\n" +
                "final_amount,\n" +
                " ts, " +
                " row_op_ts " +
                "from order_info o\n" +
                "JOIN payment_info p\n" +
                "on o.id = p.order_id");
       // paySuc.execute().print();

        // 5. 写出到 Kafka
        tEnv.executeSql("create table dwd_trade_pay_detail_suc( " +
                "user_id string , " +
                "order_id string , " +
                "payment_type string , " +
                "callback_time string , " +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));

        paySuc.executeInsert("dwd_trade_pay_detail_suc");
    }
}
/*


支付成功

让支付表和下单详情表进行 join:

字典表:
 */